package avicit.bdp.dcs.json.service.writer;

import avicit.bdp.common.utils.database.ColumnInfo;
import avicit.bdp.common.utils.database.DBUtils;
import avicit.bdp.core.constant.Constants;
import avicit.bdp.core.util.json.JSONUtils;
import avicit.bdp.dcs.datasource.dto.JobDatasource;
import avicit.bdp.dcs.datasource.service.BaseDbService;
import avicit.bdp.dcs.datasource.service.DataSourceUtils;
import avicit.bdp.dcs.desensitization.dto.DcsDesensitizationPolicyDTO;
import avicit.bdp.dcs.desensitization.dto.DcsDesensitizationRuleDTO;
import avicit.bdp.dcs.desensitization.service.DesensitizationPolicyService;
import avicit.bdp.dcs.desensitization.service.DesensitizationRuleService;
import avicit.bdp.dcs.job.dto.JobDTO;
import avicit.bdp.dcs.json.service.util.BuildJsonUtil;
import avicit.bdp.dcs.plugin.BaseDataxParam;
import avicit.bdp.dcs.plugin.rdbms.param.RdbmsDataxParam;
import avicit.bdp.dcs.semistructure.dto.HiveJsonDto;
import avicit.bdp.dcs.task.dto.ColumnMappingDto;
import avicit.bdp.dcs.task.dto.TaskDTO;
import avicit.bdp.dcs.utils.JdbcConstants;
import com.alibaba.fastjson2.JSON;
import com.google.common.collect.Lists;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
public class StructureWriterBuilder {

  @Autowired private DataSourceUtils dataSourceUtils;
  @Autowired private DesensitizationRuleService desensitizationRuleService;
  @Autowired private DesensitizationPolicyService desensitizationPolicyService;
  @Autowired private BaseDbService baseDbService;
  @Autowired private HiveWriterBuilder hiveWriterBuilder;
  @Autowired private OdpsWriterBuilder odpsWriterBuilder;
  @Autowired private BuildJsonUtil buildJsonUtil;

  /**
   * 构建结构化采集Writer
   * @param taskDTO
   * @param jobDTO
   * @return
   * @throws Exception
   */
  public BaseDataxParam buildWriterByStructure(TaskDTO taskDTO, JobDTO jobDTO) throws Exception {
    JobDatasource writerDatasource = dataSourceUtils.convertDatasource(jobDTO.getDstDatasourceId());

    // 根据列映射关系转换列名
    Map<String, String> columnMap = new HashMap<>(16);
    List<ColumnMappingDto> columnMappingDtoList =
        JSONUtils.toList(taskDTO.getMappingColumn(), ColumnMappingDto.class);
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        columnMap.put(columnMappingDto.getSrcColumnName(), columnMappingDto.getDstColumnName());
      }
    }

    // 脱敏策略
    List<ColumnMappingDto> policyList = new ArrayList<>();
    if (CollectionUtils.isNotEmpty(columnMappingDtoList)) {
      for (ColumnMappingDto columnMappingDto : columnMappingDtoList) {
        if (StringUtils.isNotEmpty(columnMappingDto.getDesensitizationRuleId())
            && StringUtils.isNotEmpty(columnMappingDto.getDesensitizationPolicyId())) {
          DcsDesensitizationRuleDTO desensitizationRuleDTO =
              desensitizationRuleService.selectByPrimaryKey(columnMappingDto.getDesensitizationRuleId());
          DcsDesensitizationPolicyDTO desensitizationPolicyDTO =
              desensitizationPolicyService.selectByPrimaryKey(columnMappingDto.getDesensitizationPolicyId());
          if (desensitizationRuleDTO != null && desensitizationPolicyDTO != null) {
            columnMappingDto.setDesensitizationRuleId(desensitizationRuleDTO.getName());
            columnMappingDto.setDesensitizationPolicyId(desensitizationPolicyDTO.getName());
            policyList.add(columnMappingDto);
          }
        }
      }
    }

    // 获取源表列名
    List<ColumnInfo> srcColumnList = baseDbService.getColumnList(jobDTO.getSrcDatasourceId(), taskDTO.getSrcTableName());

    // 构建Writer
    String dbType = writerDatasource.getDbType();
    if (JdbcConstants.HIVE.equals(dbType)) {
      HiveJsonDto hiveJsonDto = JSON.parseObject(jobDTO.getExtraParams(), HiveJsonDto.class);
      hiveJsonDto.setDstTable(taskDTO.getDstTableName());
      hiveJsonDto.setFileName(taskDTO.getDstTableName());

      // 构建HiveWriter
      return hiveWriterBuilder.buildHiveWriterByStructure(
          jobDTO, hiveJsonDto, srcColumnList, columnMap, columnMappingDtoList, writerDatasource);
    } else if (JdbcConstants.ODPS.equals(dbType)) {
      return odpsWriterBuilder.buildOdpsWriterByStructure(taskDTO, jobDTO);
    } else {
      String schema = dataSourceUtils.getDbSchema(writerDatasource);
      String tableName = DBUtils.convertTableName(writerDatasource.getDbType(), taskDTO.getDstTableName());
      tableName = dataSourceUtils.appendSchema2TableName(writerDatasource.getDbType(), tableName, schema);

      // 构建RdbmsWriter
      RdbmsDataxParam dataxParam = new RdbmsDataxParam();
      dataxParam.setTableList(Lists.newArrayList(tableName));
      dataxParam.setColumnList(buildJsonUtil.getWriterColumns(taskDTO, writerDatasource.getDbType()));
      dataxParam.setPreSql(Constants.STRING_BLANK);

      dataxParam.setPolicyList(policyList);
      return dataxParam;
    }
  }

}
